package top.lywivan.netty.reactor.cnxn;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

public class IOWorkerService {


    private final String threadNamePrefix;
    private final int numWorkerThreads;

    private ExecutorService ioWorkerPool ;

    private volatile boolean stopped = true;

    public IOWorkerService(int numWorkerThreads, String namePrefix) {
        this.threadNamePrefix = (namePrefix == null ? "" : namePrefix) + "Thread";
        this.numWorkerThreads = numWorkerThreads;
        ioWorkerPool = Executors.newFixedThreadPool(numWorkerThreads,new IOWorkThreadFactory("IOWorker"));
        stopped = false;
    }

    // 处理 IOWork
    public void schedule( IOWork ioWork) {
        if (stopped) {
            ioWork.cleanup();
            return;
        }
        // 将IOWork 包装成一个handler
        IOWorkHandler ioWorkHandler = new IOWorkHandler(ioWork);
        if (ioWorkerPool != null ) {
            try {
                ioWorkerPool.execute(ioWorkHandler);
            } catch (RejectedExecutionException e) {
                e.printStackTrace();
                ioWork.cleanup();
            }
        }else {
            ioWorkHandler.run();
        }
    }


    public abstract static class IOWork {

        /**
         * 执行 IO任务
         */
        public abstract void doWork() throws Exception;

        /**
         * (Optional) If implemented, is called if the service is stopped
         * or unable to schedule the request.
         */
        public void cleanup() {

        }

    }

    private class IOWorkHandler implements  Runnable {

        private final IOWork ioWork;

        public IOWorkHandler(IOWork ioWork) {
            this.ioWork = ioWork;
        }
        @Override
        public void run() {
            try {
                if (stopped) {
                    ioWork.cleanup();
                    return;
                }
                //处理IO
                ioWork.doWork();
            } catch (Exception e) {
                e.printStackTrace();
                ioWork.cleanup();
            }
        }
    }



    private static class IOWorkThreadFactory implements ThreadFactory {

        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        public IOWorkThreadFactory(String namePrefix) {
            this.namePrefix = namePrefix;
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
        }
        @Override
        public Thread newThread(Runnable r) {
            Thread ioThread = new Thread(group,r,namePrefix+"-"+threadNumber.getAndIncrement());
            if (!ioThread.isDaemon()) {
                ioThread.setDaemon(true);
            }
            if (ioThread.getPriority() != Thread.NORM_PRIORITY) {
                ioThread.setPriority(Thread.NORM_PRIORITY);
            }
            return ioThread;
        }
    }
}
